package org.budo.redis.jdbc.driver.executor.socket;

import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import org.budo.jdbc.driver.JdbcUrl;
import org.budo.redis.jdbc.driver.statement.hash.HDEL;
import org.budo.redis.jdbc.driver.statement.hash.HEXISTS;
import org.budo.redis.jdbc.driver.statement.hash.HGET;
import org.budo.redis.jdbc.driver.statement.hash.HGETALL;
import org.budo.redis.jdbc.driver.statement.hash.HINCRBY;
import org.budo.redis.jdbc.driver.statement.hash.HINCRBYFLOAT;
import org.budo.redis.jdbc.driver.statement.hash.HKEYS;
import org.budo.redis.jdbc.driver.statement.hash.HLEN;
import org.budo.redis.jdbc.driver.statement.hash.HMGET;
import org.budo.redis.jdbc.driver.statement.hash.HMSET;
import org.budo.redis.jdbc.driver.statement.hash.HSET;
import org.budo.redis.jdbc.driver.statement.hash.HSETNX;
import org.budo.redis.jdbc.driver.statement.hash.HVALS;
import org.budo.redis.jdbc.driver.statement.key.DEL;
import org.budo.redis.jdbc.driver.statement.key.EXISTS;
import org.budo.redis.jdbc.driver.statement.key.EXPIRE;
import org.budo.redis.jdbc.driver.statement.key.EXPIREAT;
import org.budo.redis.jdbc.driver.statement.key.KEYS;
import org.budo.redis.jdbc.driver.statement.key.MOVE;
import org.budo.redis.jdbc.driver.statement.key.PERSIST;
import org.budo.redis.jdbc.driver.statement.key.PEXPIRE;
import org.budo.redis.jdbc.driver.statement.key.PEXPIREAT;
import org.budo.redis.jdbc.driver.statement.key.PTTL;
import org.budo.redis.jdbc.driver.statement.key.RANDOMKEY;
import org.budo.redis.jdbc.driver.statement.key.TTL;
import org.budo.redis.jdbc.driver.statement.key.TYPE;
import org.budo.redis.jdbc.driver.statement.list.LINDEX;
import org.budo.redis.jdbc.driver.statement.list.LINSERT;
import org.budo.redis.jdbc.driver.statement.list.LLEN;
import org.budo.redis.jdbc.driver.statement.list.LPOP;
import org.budo.redis.jdbc.driver.statement.list.LPUSH;
import org.budo.redis.jdbc.driver.statement.list.LPUSHX;
import org.budo.redis.jdbc.driver.statement.list.LRANGE;
import org.budo.redis.jdbc.driver.statement.list.LREM;
import org.budo.redis.jdbc.driver.statement.list.LTRIM;
import org.budo.redis.jdbc.driver.statement.list.RPOP;
import org.budo.redis.jdbc.driver.statement.list.RPOPLPUSH;
import org.budo.redis.jdbc.driver.statement.list.RPUSH;
import org.budo.redis.jdbc.driver.statement.list.RPUSHX;
import org.budo.redis.jdbc.driver.statement.script.EVAL;
import org.budo.redis.jdbc.driver.statement.set.SADD;
import org.budo.redis.jdbc.driver.statement.set.SCARD;
import org.budo.redis.jdbc.driver.statement.set.SDIFF;
import org.budo.redis.jdbc.driver.statement.set.SDIFFSTORE;
import org.budo.redis.jdbc.driver.statement.set.SINTER;
import org.budo.redis.jdbc.driver.statement.set.SINTERSTORE;
import org.budo.redis.jdbc.driver.statement.set.SISMEMBER;
import org.budo.redis.jdbc.driver.statement.set.SMEMBERS;
import org.budo.redis.jdbc.driver.statement.set.SMOVE;
import org.budo.redis.jdbc.driver.statement.set.SPOP;
import org.budo.redis.jdbc.driver.statement.set.SRANDMEMBER;
import org.budo.redis.jdbc.driver.statement.set.SREM;
import org.budo.redis.jdbc.driver.statement.set.SUNION;
import org.budo.redis.jdbc.driver.statement.set.SUNIONSTORE;
import org.budo.redis.jdbc.driver.statement.set.sorted.ZADD;
import org.budo.redis.jdbc.driver.statement.set.sorted.ZCARD;
import org.budo.redis.jdbc.driver.statement.set.sorted.ZCOUNT;
import org.budo.redis.jdbc.driver.statement.set.sorted.ZINCRBY;
import org.budo.redis.jdbc.driver.statement.set.sorted.ZRANK;
import org.budo.redis.jdbc.driver.statement.set.sorted.ZREM;
import org.budo.redis.jdbc.driver.statement.set.sorted.ZREMRANGEBYRANK;
import org.budo.redis.jdbc.driver.statement.set.sorted.ZREMRANGEBYSCORE;
import org.budo.redis.jdbc.driver.statement.set.sorted.ZREVRANK;
import org.budo.redis.jdbc.driver.statement.set.sorted.ZSCORE;
import org.budo.redis.jdbc.driver.statement.string.APPEND;
import org.budo.redis.jdbc.driver.statement.string.DECR;
import org.budo.redis.jdbc.driver.statement.string.DECRBY;
import org.budo.redis.jdbc.driver.statement.string.GET;
import org.budo.redis.jdbc.driver.statement.string.GETRANGE;
import org.budo.redis.jdbc.driver.statement.string.GETSET;
import org.budo.redis.jdbc.driver.statement.string.INCR;
import org.budo.redis.jdbc.driver.statement.string.INCRBY;
import org.budo.redis.jdbc.driver.statement.string.INCRBYFLOAT;
import org.budo.redis.jdbc.driver.statement.string.MGET;
import org.budo.redis.jdbc.driver.statement.string.MSET;
import org.budo.redis.jdbc.driver.statement.string.MSETNX;
import org.budo.redis.jdbc.driver.statement.string.PSETEX;
import org.budo.redis.jdbc.driver.statement.string.SET;
import org.budo.redis.jdbc.driver.statement.string.SETEX;
import org.budo.redis.jdbc.driver.statement.string.SETNX;
import org.budo.redis.jdbc.driver.statement.string.SETRANGE;
import org.budo.redis.jdbc.driver.statement.string.STRLEN;
import org.budo.support.lang.util.BooleanUtil;
import org.budo.support.lang.util.ByteArray;
import org.budo.support.lang.util.IoUtil;
import org.budo.support.lang.util.MapUtil;
import org.budo.support.lang.util.NumberUtil;

import lombok.NoArgsConstructor;

/**
 * @author lmw
 */
@NoArgsConstructor
public class BudoSocketRedisStatementExecutor extends AbstractSocketRedisStatementExecutor {
    public BudoSocketRedisStatementExecutor(JdbcUrl jdbcUrl, String password) {
        this.setJdbcUrl(jdbcUrl);
        this.setPassword(password);
    }

    @Override
    public Long append(byte[] key, byte[] value) {
        return NumberUtil.toLong(this.request(APPEND.class, key, value));
    }

    @Override
    public Long decr(byte[] key) {
        return NumberUtil.toLong(this.request(DECR.class, key));
    }

    @Override
    public Long decrBy(byte[] key, Long value) {
        return NumberUtil.toLong(this.request(DECRBY.class, key, value.toString().getBytes()));
    }

    @Override
    public Long del(byte[][] keys) {
        return NumberUtil.toLong(this.request(DEL.class, keys));
    }

    @Override
    public Boolean exists(byte[] key) {
        return BooleanUtil.toBoolean(this.request(EXISTS.class, key));
    }

    @Override
    public Long expire(byte[] key, Integer expireIn) {
        return NumberUtil.toLong(this.request(EXPIRE.class, key, expireIn.toString().getBytes()));
    }

    @Override
    public Long expireAt(byte[] key, Long unixTime) {
        return NumberUtil.toLong(this.request(EXPIREAT.class, key, unixTime.toString().getBytes()));
    }

    @Override
    public byte[] get(byte[] key) {
        return (byte[]) this.request(GET.class, key);
    }

    @Override
    public byte[] getrange(byte[] key, Long from, Long to) {
        return (byte[]) this.request(GETRANGE.class, key, from.toString().getBytes(), to.toString().getBytes());
    }

    @Override
    public byte[] getSet(byte[] key, byte[] value) {
        return (byte[]) this.request(GETSET.class, key, value);
    }

    @Override
    public Long hdel(byte[] key, byte[][] fields) {
        ByteArray byteArray = new ByteArray() //
                .write(HDEL.class.getSimpleName().getBytes()) //
                .write(SPACE_BYTES) //
                .write(key) //
                .write(SPACE_BYTES) //
                .writeJoin(fields, SPACE_BYTES) //
                .write(SPACE_BYTES) //
                .write(NOFIELD_BYTES) //
                .write(CRLF_BYTES);
        byte[] command = byteArray.getBytes();

        return NumberUtil.toLong(this.request(command));
    }

    @Override
    public Boolean hexists(byte[] key, byte[] field) {
        return BooleanUtil.toBoolean(this.request(HEXISTS.class, key, field));
    }

    @Override
    public byte[] hget(byte[] key, byte[] field) {
        return (byte[]) this.request(HGET.class, key, field);
    }

    @Override
    public Map<byte[], byte[]> hgetAll(byte[] key) {
        byte[][] values = (byte[][]) this.request(HGETALL.class, key);
        return MapUtil.newMap((Object[]) values);
    }

    @Override
    public Long hincrBy(byte[] key, byte[] field, Long value) {
        return NumberUtil.toLong(this.request(HINCRBY.class, key, field, value.toString().getBytes()));
    }

    @Override
    public Double hincrByFloat(byte[] key, byte[] field, Double value) {
        return NumberUtil.toDouble(this.request(HINCRBYFLOAT.class, key, field, value.toString().getBytes()));
    }

    @Override
    public byte[][] hkeys(byte[] key) {
        return (byte[][]) this.request(HKEYS.class, key);
    }

    @Override
    public Long hlen(byte[] key) {
        return NumberUtil.toLong(this.request(HLEN.class, key));
    }

    @Override
    public byte[][] hmget(byte[] key, byte[][] fields) {
        ByteArray byteArray = new ByteArray() //
                .write(HMGET.class.getSimpleName().getBytes()).write(SPACE_BYTES) //
                .write(key).write(SPACE_BYTES) //
                .writeJoin(fields, SPACE_BYTES) //
                .write(SPACE_BYTES) //
                .write(NOFIELD_BYTES) //
                .write(CRLF_BYTES);
        byte[] command = byteArray.getBytes();

        return (byte[][]) this.request(command);
    }

    @Override
    public String hmset(byte[] key, Map<byte[], byte[]> hash) {
        ByteArray byteArray = new ByteArray() //
                .write(HMSET.class.getSimpleName().getBytes()).write(SPACE_BYTES) //
                .write(key);

        Set<Entry<byte[], byte[]>> entrySet = hash.entrySet();
        for (Entry<byte[], byte[]> entry : entrySet) {
            byteArray.write(SPACE_BYTES) //
                    .write(entry.getKey()) //
                    .write(SPACE_BYTES) //
                    .write(QUOT_BYTES) //
                    .write(entry.getValue()) //
                    .write(QUOT_BYTES);
        }

        byteArray.write(CRLF_BYTES);

        byte[] command = byteArray.getBytes();
        return new String((byte[]) this.request(command));
    }

    @Override
    public Long hset(byte[] key, byte[] field, byte[] value) {
        return NumberUtil.toLong(this.request(HSET.class, key, field, value));
    }

    @Override
    public Long hsetnx(byte[] key, byte[] field, byte[] value) {
        return NumberUtil.toLong(this.request(HSETNX.class, key, field, value));
    }

    @Override
    public byte[][] hvals(byte[] key) {
        return (byte[][]) this.request(HVALS.class, key);
    }

    @Override
    public Long incr(byte[] key) {
        return NumberUtil.toLong(this.request(INCR.class, key));
    }

    @Override
    public Long incrBy(byte[] key, Long value) {
        return NumberUtil.toLong(this.request(INCRBY.class, key, value.toString().getBytes()));
    }

    @Override
    public Double incrByFloat(byte[] key, Double value) {
        return NumberUtil.toDouble(this.request(INCRBYFLOAT.class, key, value.toString().getBytes()));
    }

    @Override
    public byte[][] keys(byte[] pattern) {
        return (byte[][]) this.request(KEYS.class, pattern);
    }

    @Override
    public byte[] lindex(byte[] key, Long index) {
        return (byte[]) this.request(LINDEX.class, key, index.toString().getBytes());
    }

    @Override
    public Long linsert(byte[] key, byte[] position, byte[] pivot, byte[] value) {
        return NumberUtil.toLong(this.request(LINSERT.class, key, position, pivot, value));
    }

    @Override
    public Long llen(byte[] key) {
        return NumberUtil.toLong(this.request(LLEN.class, key));
    }

    @Override
    public byte[] lpop(byte[] key) {
        return (byte[]) this.request(LPOP.class, key);
    }

    @Override
    public Long lpush(byte[] key, byte[][] values) {
        ByteArray byteArray = new ByteArray() //
                .write(LPUSH.class.getSimpleName().getBytes())//
                .write(SPACE_BYTES)//
                .write(key)//
                .write(SPACE_BYTES) //
                .writeJoin(values, SPACE_BYTES) //
                .write(CRLF_BYTES);
        byte[] command = byteArray.getBytes();
        return NumberUtil.toLong(this.request(command));
    }

    @Override
    public Long lpushx(byte[] key, byte[] value) {
        return NumberUtil.toLong(this.request(LPUSHX.class, key, value));
    }

    @Override
    public byte[][] lrange(byte[] key, Long start, Long end) {
        return (byte[][]) this.request(LRANGE.class, key, start.toString().getBytes(), end.toString().getBytes());
    }

    @Override
    public Long lrem(byte[] key, Long count, byte[] value) {
        return NumberUtil.toLong(this.request(LREM.class, key, count.toString().getBytes(), value));
    }

    @Override
    public String lset(byte[] key, Long index, byte[] value) {
        throw new RuntimeException();
    }

    @Override
    public String ltrim(byte[] key, Long start, Long end) {
        return IoUtil.toString(this.request(LTRIM.class, key, start.toString().getBytes(), end.toString().getBytes()));
    }

    @Override
    public byte[][] mget(byte[][] keys) {
        return (byte[][]) this.request(MGET.class, keys);
    }

    @Override
    public Long move(byte[] key, Integer dbIndex) {
        return NumberUtil.toLong(this.request(MOVE.class, key, dbIndex.toString().getBytes()));
    }

    @Override
    public String mset(byte[][] keyValues) {
        return IoUtil.toString(this.request(MSET.class, keyValues));
    }

    @Override
    public Long msetnx(byte[][] keyValues) {
        return NumberUtil.toLong(this.request(MSETNX.class, keyValues));
    }

    @Override
    public Long persist(byte[] key) {
        return NumberUtil.toLong(this.request(PERSIST.class, key));
    }

    @Override
    public Long pexpire(byte[] key, Long milliSeconds) {
        return NumberUtil.toLong(this.request(PEXPIRE.class, key, milliSeconds.toString().getBytes()));
    }

    @Override
    public Long pexpireAt(byte[] key, Long milliSecondsTimestamp) {
        return NumberUtil.toLong(this.request(PEXPIREAT.class, key, milliSecondsTimestamp.toString().getBytes()));
    }

    @Override
    public String psetex(byte[] key, Long milliSeconds, byte[] value) {
        return IoUtil.toString(this.request(PSETEX.class, key, milliSeconds.toString().getBytes(), value));
    }

    @Override
    public Long pttl(byte[] key) {
        return NumberUtil.toLong(this.request(PTTL.class, key));
    }

    @Override
    public String randomKey() {
        return IoUtil.toString(this.request(RANDOMKEY.class));
    }

    @Override
    public String rename(byte[] key, byte[] newKey) {
        throw new RuntimeException();
    }

    @Override
    public Long renamenx(byte[] key, byte[] newKey) {
        throw new RuntimeException();
    }

    @Override
    public byte[] rpop(byte[] key) {
        return (byte[]) this.request(RPOP.class, key);
    }

    @Override
    public byte[] rpoplpush(byte[] srcKey, byte[] destKey) {
        return (byte[]) this.request(RPOPLPUSH.class, srcKey, destKey);
    }

    @Override
    public Long rpush(byte[] key, byte[][] values) {
        ByteArray byteArray = new ByteArray() //
                .write(RPUSH.class.getSimpleName().getBytes()) //
                .write(SPACE_BYTES) //
                .write(key) //
                .write(SPACE_BYTES) //
                .writeJoin(values, SPACE_BYTES) //
                .write(CRLF_BYTES);
        byte[] command = byteArray.getBytes();
        return NumberUtil.toLong(this.request(command));
    }

    @Override
    public Long rpushx(byte[] key, byte[] value) {
        return NumberUtil.toLong(this.request(RPUSHX.class, key, value));
    }

    @Override
    public Long sadd(byte[] key, byte[][] values) {
        ByteArray byteArray = new ByteArray() //
                .write(SADD.class.getSimpleName().getBytes()) //
                .write(SPACE_BYTES) //
                .write(key) //
                .write(SPACE_BYTES) //
                .writeJoin(values, SPACE_BYTES) //
                .write(CRLF_BYTES);
        byte[] command = byteArray.getBytes();
        return NumberUtil.toLong(this.request(command));
    }

    @Override
    public Long scard(byte[] key) {
        return NumberUtil.toLong(this.request(SCARD.class, key));
    }

    @Override
    public byte[][] sdiff(byte[][] keys) {
        return (byte[][]) this.request(SDIFF.class, keys);
    }

    @Override
    public Long sdiffstore(byte[] destKey, byte[][] keys) {
        ByteArray byteArray = new ByteArray() //
                .write(SDIFFSTORE.class.getSimpleName().getBytes()) //
                .write(SPACE_BYTES) //
                .write(destKey) //
                .write(SPACE_BYTES) //
                .writeJoin(keys, SPACE_BYTES) //
                .write(CRLF_BYTES);
        byte[] command = byteArray.getBytes();
        return NumberUtil.toLong(this.request(command));
    }

    @Override
    public String set(byte[] key, byte[] value) {
        return IoUtil.toString(this.request(SET.class, key, value));
    }

    @Override
    public String setex(byte[] key, Integer seconds, byte[] value) {
        return IoUtil.toString(this.request(SETEX.class, key, seconds.toString().getBytes(), value));
    }

    @Override
    public Long setnx(byte[] key, byte[] value) {
        return NumberUtil.toLong(this.request(SETNX.class, key, value));
    }

    @Override
    public Long setrange(byte[] key, Long offset, byte[] value) {
        return NumberUtil.toLong(this.request(SETRANGE.class, key, offset.toString().getBytes(), value));
    }

    @Override
    public byte[][] sinter(byte[][] keys) {
        return (byte[][]) this.request(SINTER.class, keys);
    }

    @Override
    public Long sinterstore(byte[] destKey, byte[][] keys) {
        ByteArray byteArray = new ByteArray() //
                .write(SINTERSTORE.class.getSimpleName().getBytes())//
                .write(SPACE_BYTES)//
                .write(destKey)//
                .write(SPACE_BYTES)//
                .writeJoin(keys, SPACE_BYTES) //
                .write(CRLF_BYTES);
        byte[] command = byteArray.getBytes();
        return NumberUtil.toLong(this.request(command));
    }

    @Override
    public Boolean sismember(byte[] key, byte[] value) {
        return BooleanUtil.toBoolean(this.request(SISMEMBER.class, key, value));
    }

    @Override
    public byte[][] smembers(byte[] key) {
        return (byte[][]) this.request(SMEMBERS.class, key);
    }

    @Override
    public Long smove(byte[] srcKey, byte[] destKey, byte[] value) {
        return NumberUtil.toLong(this.request(SMOVE.class, srcKey, destKey, value));
    }

    @Override
    public byte[] spop(byte[] key) {
        return (byte[]) this.request(SPOP.class, key);
    }

    @Override
    public byte[][] srandmember(byte[] key, Integer count) {
        byte[][] params;
        if (null == count || 0 == count) {
            params = new byte[][] { key };
        } else {
            params = new byte[][] { key, count.toString().getBytes() };
        }

        return (byte[][]) this.request(SRANDMEMBER.class, params);
    }

    @Override
    public Long srem(byte[] key, byte[][] values) {
        ByteArray byteArray = new ByteArray() //
                .write(SREM.class.getSimpleName().getBytes()) //
                .write(SPACE_BYTES) //
                .write(key) //
                .write(SPACE_BYTES) //
                .writeJoin(values, SPACE_BYTES) //
                .write(CRLF_BYTES);
        byte[] command = byteArray.getBytes();
        return NumberUtil.toLong(this.request(command));
    }

    @Override
    public Long strlen(byte[] key) {
        return NumberUtil.toLong(this.request(STRLEN.class, key));
    }

    @Override
    public byte[][] sunion(byte[][] keys) {
        return (byte[][]) this.request(SUNION.class, keys);
    }

    @Override
    public Long sunionstore(byte[] destKey, byte[][] keys) {
        ByteArray byteArray = new ByteArray() //
                .write(SUNIONSTORE.class.getSimpleName().getBytes()) //
                .write(SPACE_BYTES) //
                .write(destKey) //
                .write(SPACE_BYTES) //
                .writeJoin(keys, SPACE_BYTES) //
                .write(CRLF_BYTES);
        byte[] command = byteArray.getBytes();
        return NumberUtil.toLong(this.request(command));
    }

    @Override
    public Long ttl(byte[] key) {
        return NumberUtil.toLong(this.request(TTL.class, key));
    }

    @Override
    public String type(byte[] key) {
        return IoUtil.toString(this.request(TYPE.class, key));
    }

    @Override
    public Long zadd(byte[] key, Map<byte[], Double> members) {
        ByteArray byteArray = new ByteArray();
        byteArray.write(ZADD.class.getSimpleName().getBytes());
        byteArray.write(SPACE_BYTES);
        byteArray.write(key);

        Set<Entry<byte[], Double>> entrySet = members.entrySet();
        for (Entry<byte[], Double> entry : entrySet) {
            byteArray.write(SPACE_BYTES);
            byteArray.write(entry.getValue().toString().getBytes());
            byteArray.write(SPACE_BYTES);
            byteArray.write(QUOT_BYTES);
            byteArray.write(entry.getKey());
            byteArray.write(QUOT_BYTES);
        }

        byteArray.write(CRLF_BYTES);
        byte[] command = byteArray.getBytes();
        return NumberUtil.toLong(this.request(command));
    }

    @Override
    public Long zcard(byte[] key) {
        return NumberUtil.toLong(this.request(ZCARD.class, key));
    }

    @Override
    public Long zcount(byte[] key, Double min, Double max) {
        return NumberUtil.toLong(this.request(ZCOUNT.class, key, min.toString().getBytes(), max.toString().getBytes()));
    }

    @Override
    public Double zincrby(byte[] key, Double score, byte[] value) {
        return NumberUtil.toDouble(this.request(ZINCRBY.class, key, score.toString().getBytes(), value));
    }

    @Override
    public Long zlexcount(byte[] key, byte[] min, byte[] max) {
        throw new RuntimeException();
    }

    @Override
    public Long zrank(byte[] key, byte[] value) {
        return NumberUtil.toLong(this.request(ZRANK.class, key, value));
    }

    @Override
    public Long zrem(byte[] key, byte[][] values) {
        ByteArray byteArray = new ByteArray() //
                .write(ZREM.class.getSimpleName().getBytes()) //
                .write(SPACE_BYTES) //
                .write(key) //
                .write(SPACE_BYTES) //
                .writeJoin(values, SPACE_BYTES) //
                .write(CRLF_BYTES);
        byte[] command = byteArray.getBytes();
        return NumberUtil.toLong(this.request(command));
    }

    @Override
    public Long zremrangeByRank(byte[] key, Long from, Long to) {
        return NumberUtil.toLong(this.request(ZREMRANGEBYRANK.class, key, from.toString().getBytes(), to.toString().getBytes()));
    }

    @Override
    public Long zremrangeByScore(byte[] key, Double min, Double max) {
        return NumberUtil.toLong(this.request(ZREMRANGEBYSCORE.class, key, min.toString().getBytes(), max.toString().getBytes()));
    }

    @Override
    public Long zrevrank(byte[] key, byte[] value) {
        return NumberUtil.toLong(this.request(ZREVRANK.class, key, value));
    }

    @Override
    public Double zscore(byte[] key, byte[] value) {
        return NumberUtil.toDouble(this.request(ZSCORE.class, key, value));
    }

    @Override
    public Object eval(byte[] script) {
        return this.request(EVAL.class, script, "0".getBytes());
    }

    @Override
    public byte[][] configGet(byte[] pattern) {
        throw new RuntimeException();
    }
}